package com.king.function

import java.sql.{Connection, DriverManager, PreparedStatement, SQLException}

import com.alibaba.fastjson.{JSON, JSONObject}
import com.king.bean.TableProcess
import com.king.config.GmallConfig
import org.apache.flink.api.common.state.MapStateDescriptor
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction
import org.apache.flink.streaming.api.scala.OutputTag
import org.apache.flink.util.Collector

/**
 * @Author: KingWang
 * @Date: 2022/1/11  
 * @Desc:
 **/
class TableProcessFunction(outputTag:OutputTag[JSONObject],mapStateDescriptor: MapStateDescriptor[String,TableProcess]) extends BroadcastProcessFunction[JSONObject,String,JSONObject]{


  var connection:Connection = _
  var prepareStatement:PreparedStatement = _

  override def open(parameters: Configuration): Unit = {
    try {
//      Class.forName(GmallConfig.PHOENIX_DRIVER)
      connection = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER)
      connection.setSchema(GmallConfig.HBASE_SCHEMA)
    }catch{
      case e:ClassNotFoundException => println("org.apache.phoenix.jdbc.PhoenixDriver未找到")
      case _:Exception => println("phoenix驱动加载异常")
    }
  }



  //处理广播流：维度表处理，广播流数据是手工指定的表的目标到kafka还是hbase
  //value: {"db":"","tn":"","before":"","after":"","type":""}
  override def processBroadcastElement(value: String, ctx: BroadcastProcessFunction[JSONObject, String, JSONObject]#Context, out: Collector[JSONObject]): Unit = {

    //1.获取并解析数据
    val jsonObject = JSON.parseObject(value)
    val data = jsonObject.getString("after")
    val tableProcess = JSON.parseObject(data,classOf[TableProcess])

    //2.建表
    if(tableProcess.SINK_TYPE_HBASE.equals(tableProcess.sinkType)){
      checkTable(tableProcess.sinkTable,
        tableProcess.sinkColumns,
        tableProcess.sinkPk,
        tableProcess.sinkExtend)
    }

    //3.写入状态,广播出去
    val broadcastState = ctx.getBroadcastState(mapStateDescriptor)
    val key = tableProcess.sinkTable + "-" + tableProcess.sinkType
    broadcastState.put(key, tableProcess)



  }



  //处理主流数据： 主流数据是实际的业务数据
  //value: {"db":"","tn":"","before":"","after":"","type":""}
  override def processElement(value: JSONObject, ctx: BroadcastProcessFunction[JSONObject, String, JSONObject]#ReadOnlyContext, out: Collector[JSONObject]): Unit = {

    //1.获取状态数据
    val broadcastState = ctx.getBroadcastState(mapStateDescriptor)
    val key = value.getString("tableName") + "-" + value.getString("type")
    val tableProcess = broadcastState.get(key)
    if(null != tableProcess){

      //2. 过滤字段，将指定查询的字段获取对应的值，保存到newValue[JSONObject]
      val oldValue = value.getJSONObject("after")
      val queryFields = tableProcess.sinkColumns.split(",")
      val newValue = new JSONObject()
      queryFields.map{field =>
        newValue.put(field, oldValue.get(field))
      }

      //3. 分流到kafka或者Hbase
      //将输出表/主题的信息写入Value
      val sinkType = tableProcess.sinkType
      newValue.put("sinkTable",tableProcess.sinkTable)

      if(tableProcess.SINK_TYPE_KAFKA.equals(sinkType)){
        //kafka数据，写入主流
        out.collect(newValue)
      }else if(tableProcess.SINK_TYPE_HBASE.equals(sinkType)){
        ctx.output[JSONObject](outputTag, newValue)
      }


    }else{
      println("该组合key：" + key +" 不存在")
    }


  }


  //判断phoenix中表是否存在，如果不存在，则自动创建该表
  //建表语句：create table if not exists mydb.test(id varchar primary key, name varchar, sex varchar)
  def checkTable(sinkTable: String, sinkColumns: String, sinkPk: String, sinkExtend: String) = {

    try {

      val database = GmallConfig.HBASE_SCHEMA
      val begin_sql =
        s"""
           |create table if not exists $database.$sinkTable (
           |""".stripMargin

      val pk = if (null != sinkPk) sinkPk else "id"
      val extend = if (null != sinkExtend) sinkExtend else ""
      val end_sql = sinkColumns.split(",")
        .map(f => if (f.equals(pk)) f.concat(" varchar primary key") else f.concat(" varchar "))
        .mkString(",")
        .concat(")")
        .concat(extend)

      val create_table_sql = begin_sql + end_sql
      println(s"创建表 $database.$sinkTable 语句如下：")
      println(create_table_sql)

      //预编译SQL
      prepareStatement = connection.prepareStatement(create_table_sql)
      prepareStatement.execute()

    }catch {
      case e:SQLException => throw new RuntimeException("Phoenix表" + sinkTable + "建表失败")
    }finally {
      try{
        if(null != prepareStatement) prepareStatement.close()
      }catch {
        case e:SQLException => e.printStackTrace()
      }
    }

  }

  override def close(): Unit = {
    try{
//      if(null != prepareStatement) prepareStatement.close()
      if(null != connection) connection.close()
    }catch {
      case e:SQLException => e.printStackTrace()
    }
  }


}
